package com.atguigu.qzpoint.producer

import com.atguigu.constan.Constan
import com.atguigu.util.{ConfigManager, KafkaUtil}
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.spark.{SparkConf, SparkContext}

/**
 * description ：注册数据写入kafka
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/3/11 14:36
 * modified By ：
 * version:    : 1.0
 */
object RegisterProducer {

  def main(args: Array[String]): Unit = {
    // 获取 sparkContext
    val conf = new SparkConf().setAppName(this.getClass.getName).setMaster("local[*]")
    val sc = new SparkContext(conf)

    // 读取文件
    val source = sc.textFile(s"${Constan.WINDWO_FILE_PATH}\\register.log", 10)
    source.foreachPartition(it => {
      // 获取 kafka生产者
      val producer = KafkaUtil.getKafkaProducer(ConfigManager.getProperty("kafka.bootstrap.servers"))

      // 遍历第一行数据并写入 kafka 中
      it.foreach(line => {
        val msg = new ProducerRecord[String, String](Constan.KAFKA_TOPIC_REGISTER, line)
        producer.send(msg)
      })

      // 关闭 kafka 生产者
      producer.flush()
      producer.close()
    })
  }


}
